#include "common.h"
#include "config.h"

#define TEST_NZ(x) do { if ( (x)) die("error: " #x " failed (returned non-zero)." ); } while (0)
#define TEST_Z(x)  do { if (!(x)) die("error: " #x " failed (returned zero/null)."); } while (0)

DataNode* datanodes=NULL;
struct rdma_event_channel *event_channel = NULL;
struct Context *s_ctx = NULL;
Connection** conn2datanodes=NULL;
int** block_loc=NULL;
long requestid=0;
pthread_cond_t cond=PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
int flag=1;
long start_time,end_time,s_time;
long write_total_time=0;
long read_total_time=0;
long total_time=0;
FILE* latency_file=NULL;
//=========================================================
void* conn_event(void* arg);
void* poll_cq(void *ctx);
void build_context(struct ibv_context *verbs);
void build_qp_attr(struct ibv_qp_init_attr *qp_attr);
int on_addr_resolved(struct rdma_cm_id *id);
int on_route_resolved(struct rdma_cm_id *id);
int on_connection(struct rdma_cm_id *id);
int on_disconnect(struct rdma_cm_id *id);
void destroy_connection(void *context);
int on_event(struct rdma_cm_event *event);
void register_memory(Connection *conn);
void on_completion(struct ibv_wc *wc);
void build_params(struct rdma_conn_param *params);
void init_block_loc();
void play_trace_msr(char* tracepath);
void play_trace_ycsb(char* tracepath);
void exeWritetrace(WriteRequest* request);
void exeReadtrace(ReadRequest* request);
void send_WriteRequest(WriteRequest* request);
void send_ReadRequest(ReadRequest* request);
void post_receives4write(WriteRequest* request);
void post_receives4read(ReadRequest* request);
void send_UpdateMessage(int fd,UpdateMessage* message);
void recv_UpdateMessage(int fd,UpdateMessage* message);
void reconstruction(int recovery_id,int blocknum);

int main(int argc, char const *argv[])
{

	if (argc!=2)
	{
		printf("please input ./client latency_file_num\n");
		return 0;
	}

	char* filename = (char*)malloc(sizeof(char)*30);
	sprintf(filename,"latency%s",argv[1]);
	latency_file = fopen(filename,"a"); 

	datanodes = read_config_Node(NODE_CONFIG);
	init_block_loc();
	conn2datanodes = (Connection**)malloc(sizeof(Connection*)*DATANODE_NUM);


	struct addrinfo *addr;
	struct rdma_cm_id *conn_id= NULL;
	TEST_Z(event_channel = rdma_create_event_channel());
	
	for (int i = 0; i < DATANODE_NUM; ++i)
	{
		TEST_NZ(rdma_create_id(event_channel, &conn_id, NULL, RDMA_PS_TCP));
		char* port = (char*)malloc(20);
		sprintf(port,"%d",RDMA_PORT); 
		char* target_ip=datanodes[i].rdma_ip;
		TEST_NZ(getaddrinfo(target_ip, port,NULL, &addr));
		TEST_NZ(rdma_resolve_addr(conn_id, NULL, addr->ai_addr, TIMEOUT_IN_MS));
		free(port);
	}
	freeaddrinfo(addr);

	pthread_t threadid4conn_event;
	pthread_create(&threadid4conn_event,NULL,conn_event,NULL);


  	sleep(10);
	//播放trace
	printf("starting trace playing.........\n");
	if (isYCSB==1)
	{
		play_trace_ycsb(YCSB_TRACE);	
	}else{
		play_trace_msr(MSR_TRACE);
		if (IS_RECOVERY==1)
		{
			reconstruction(RECOVERY_ID,RECOVERY_NUM);	
		}
	}
	
	printf("finishing trace playing.........\n");
	fclose(latency_file);
  	return 0;	
}

void play_trace_ycsb(char* tracepath)
{
	FILE* fp;
	if ((fp=fopen(tracepath,"r"))==NULL)
	{
		printf("open trace failed in play_trace_ycsb\n");
		exit(0);
	}

	int ret;
	char operation[150];
	char blockid[20];
	char op_type[4];
	char divider=',';
	long write_count=0;
	long read_count=0;
	int blockid_int=0;

	while(fgets(operation,sizeof(operation),fp)){
		new_strtok(operation,divider,op_type);
		new_strtok(operation,divider,blockid);

		sscanf(blockid,"%d",&blockid_int);

		if ((ret = strcmp("U",op_type))==0)
		{
			write_count++;
			if (write_count%500==0)
			{
				printf("%ld write request finish \n",write_count);
			}
			start_time = getcurrent_time();

			int stripeid = blockid_int/K;
			int offsetid = blockid_int%K;
			int targetid = block_loc[stripeid][offsetid];
			WriteRequest* request = (WriteRequest*)malloc(sizeof(WriteRequest));
			request->type=1;
			request->requestid=++requestid;
			request->targetid=targetid;
			for (int j = 0; j < R; j++) {
				request->paritynodeid[j] = block_loc[stripeid][K+j];
				request->parityid[j] = stripeid*(K+R)+K+j;
			}
			request->blockid=stripeid*(K+R)+offsetid;
			gene_radm_buff(request->newdata,BLK_SIZE);
			exeWritetrace(request);
			free(request);

			end_time = getcurrent_time();
			s_time = end_time - start_time;
			printf("%ld write request latency:%ld mius\n",write_count,s_time);
			fprintf(latency_file, "%s,%ld\n",op_type,s_time);
			write_total_time=write_total_time+s_time;
			total_time=total_time+s_time;

		}else if ((ret=strcmp("R",op_type))==0){
			read_count++;
			if (read_count%500==0)
			{
				printf("%ld read request finish \n",read_count);
			}
			start_time = getcurrent_time();

			int stripeid = blockid_int/K;
			int offsetid = blockid_int%K;
			int targetid = block_loc[stripeid][offsetid];
			ReadRequest* request = (ReadRequest*)malloc(sizeof(ReadRequest));
			request->type=0;
			request->requestid=++requestid;
			request->targetid=targetid;
			request->blockid = stripeid*(K+R)+offsetid;
			exeReadtrace(request);
			free(request);

			end_time = getcurrent_time();
			s_time = end_time - start_time;
			printf("%ld read request latency:%ld mius\n",read_count,s_time);
			fprintf(latency_file, "%s,%ld\n",op_type,s_time);
			read_total_time=read_total_time+s_time;
			total_time=total_time+s_time;

		}else if((ret=strcmp("W",op_type))==0){
			printf("Write is not support now op_type: %s\n",op_type);
			continue;
		}else if((ret=strcmp("D",op_type))==0){
			printf("Delete is not support now op_type: %s\n",op_type);
			continue;
		}else{
			printf("unknown op_type: %s\n",op_type);
			continue;
		}
	}

	printf("write_count:%ld\n",write_count);
	float write_latency = write_total_time*1.0/write_count;
	printf("write_latency:%f mius\n",write_latency);

	printf("read_count:%ld\n",read_count);
	float read_latency = read_total_time*1.0/read_count;
	printf("read_latency:%f mius\n",read_latency);

	printf("total_count:%ld\n",read_count+write_count);
	float latency = total_time*1.0/(read_count+write_count);
	printf("latency:%f mius\n",latency);

	fclose(fp);

}

void play_trace_msr(char* tracepath)
{
	FILE* fp;
	if ((fp=fopen(tracepath,"r"))==NULL)
	{
		printf("open trace failed in play_trace_msr\n");
		exit(0);
	}

	char operation[150];
	char time_stamp[50];
	char workload_name[10];
	char volumn_id[5];
	char op_type[10];
	char offset[20];
	char size[10];
	char usertime[10];
	char divider=',';

	int i;
	int access_start_block,access_end_block;
	int ret;
	long write_count;
	long read_count;

	long long *size_int;
	long long *offset_int;
	long long a,b;
	a=0LL;
	b=0LL;
	offset_int=&a;
	size_int=&b;
	write_count=0;
	read_count=0;

	while(fgets(operation,sizeof(operation),fp)){
		new_strtok(operation,divider,time_stamp);
		new_strtok(operation,divider,workload_name);
		new_strtok(operation,divider,volumn_id);
		new_strtok(operation,divider,op_type);
		new_strtok(operation,divider,offset);
		new_strtok(operation,divider,size);
		new_strtok(operation,divider,usertime);

		if ((ret=strcmp(op_type,"Read"))==0)
		{
			read_count++;
			if (read_count%500==0)
			{
				printf("%ld read request finish \n",read_count);
			}

			trnsfm_char_to_int(offset,offset_int);
			trnsfm_char_to_int(size,size_int);
			access_start_block=(*offset_int)/((long long)(DUMMY_BLK_SIZE));
			access_end_block=(*offset_int+*size_int-1)/((long long)(DUMMY_BLK_SIZE));

			start_time = getcurrent_time();
			for (i = access_start_block; i <= access_end_block; ++i)
			{
				//发送write请求给相应的datanode
				int stripeid = i/K;
				int offsetid = i%K;
				int targetid = block_loc[stripeid][offsetid];
				//发送读数据块请求
				ReadRequest* request = (ReadRequest*)malloc(sizeof(ReadRequest));
				request->type=0;
				request->requestid=++requestid;
				request->targetid=targetid;
				request->blockid = stripeid*(K+R)+offsetid;
				exeReadtrace(request);
				free(request);	
			}
			end_time = getcurrent_time();
			s_time = end_time - start_time;
			printf("%ld read request latency:%ld mius\n",read_count,s_time);
			fprintf(latency_file, "%s,%ld\n",op_type,s_time);
			read_total_time=read_total_time+s_time;
			total_time=total_time+s_time;
		
		}else if ((ret=strcmp(op_type,"Write"))==0)
		{
			write_count++;
			if (write_count%500==0)
			{
				printf("%ld write request finish \n",write_count);
			}

			trnsfm_char_to_int(offset,offset_int);
			trnsfm_char_to_int(size,size_int);
			access_start_block=(*offset_int)/((long long)(DUMMY_BLK_SIZE));
			access_end_block=(*offset_int+*size_int-1)/((long long)(DUMMY_BLK_SIZE));

			start_time = getcurrent_time();
			for (i = access_start_block; i <= access_end_block; ++i)
			{
				//发送write请求给相应的datanode
				int stripeid = i/K;
				int offsetid = i%K;
				int targetid = block_loc[stripeid][offsetid];
				WriteRequest* request = (WriteRequest*)malloc(sizeof(WriteRequest));
				request->type=1;
				request->requestid=++requestid;
				request->targetid=targetid;
				for (int j = 0; j < R; j++) {
					request->paritynodeid[j] = block_loc[stripeid][K+j];
					request->parityid[j] = stripeid*(K+R)+K+j;
				}
				request->blockid=stripeid*(K+R)+offsetid;
	      		//printf("targetid:%d,paritynodeid1:%d,paritynodeid2:%d\n",request->targetid,request->paritynodeid[0],request->paritynodeid[1]);
				gene_radm_buff(request->newdata,BLK_SIZE);
				exeWritetrace(request);
				free(request);
			}
			end_time = getcurrent_time();
			s_time = end_time - start_time;
			printf("%ld write request latency:%ld mius\n",write_count,s_time);
			fprintf(latency_file, "%s,%ld\n",op_type,s_time);
			write_total_time=write_total_time+s_time;
			total_time=total_time+s_time;
		}else{
			printf("unknown op_type: %s\n",op_type);
			continue;
		}
		
	}

	printf("write_count:%ld\n",write_count);
	//printf("write_total_time:%ld\n",write_total_time);
	//float write_thoughtput = write_count*1.0/write_total_time*1000*1000;
	//printf("write_thoughtput:%f ops\n",write_thoughtput);
	float write_latency = write_total_time*1.0/write_count;
	printf("write_latency:%f mius\n",write_latency);

	printf("read_count:%ld\n",read_count);
	//printf("read_total_time:%ld\n",read_total_time);
	//float read_thoughtput = read_count*1.0/read_total_time*1000*1000;
	//printf("read_thoughtput:%f ops\n",read_thoughtput);
	float read_latency = read_total_time*1.0/read_count;
	printf("read_latency:%f mius\n",read_latency);

	printf("total_count:%ld\n",read_count+write_count);
	//printf("total_time:%ld\n",total_time);
	//float thoughtput = (read_count+write_count)*1.0/total_time*1000*1000;
	//printf("thoughtput:%f ops\n",thoughtput);
	float latency = total_time*1.0/(read_count+write_count);
	printf("latency:%f mius\n",latency);

}

void exeReadtrace(ReadRequest* request)
{
	printf("start read_request %ld\n",request->requestid);
	post_receives4read(request);
	send_ReadRequest(request);
	pthread_mutex_lock(&mutex);
	while(flag==1){
		pthread_cond_wait(&cond,&mutex);
	}
	flag=1;
	pthread_mutex_unlock(&mutex);
	printf("finish read_request %ld\n",request->requestid);
}

void exeWritetrace(WriteRequest* request)
{
  printf("start write_request %ld\n",request->requestid);
	post_receives4write(request);
  	send_WriteRequest(request);
	pthread_mutex_lock(&mutex);
	while(flag==1){
		pthread_cond_wait(&cond,&mutex);
	}
	flag=1;
	pthread_mutex_unlock(&mutex);
  	printf("finish write_request %ld\n",request->requestid);
}

void send_WriteRequest(WriteRequest* request)
{
	struct Connection* conn = conn2datanodes[request->targetid];
	struct ibv_send_wr wr, *bad_wr = NULL;
  	struct ibv_sge sge;
  	Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

  	memcpy(client_conn->send_region,request,sizeof(WriteRequest));

  	memset(&wr, 0, sizeof(wr));

	wr.wr_id = (uintptr_t)conn;
	wr.opcode = IBV_WR_SEND;
	wr.sg_list = &sge;
	wr.num_sge = 1;
	wr.send_flags = IBV_SEND_SIGNALED;

	sge.addr = (uintptr_t)client_conn->send_region;
	sge.length = sizeof(WriteRequest);
	sge.lkey = client_conn->send_mr->lkey;

	TEST_NZ(ibv_post_send(conn->qp, &wr, &bad_wr));
}

void send_ReadRequest(ReadRequest* request)
{
	struct Connection* conn = conn2datanodes[request->targetid];
	struct ibv_send_wr wr, *bad_wr = NULL;
  	struct ibv_sge sge;
  	Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

  	memcpy(client_conn->send_region,request,sizeof(ReadRequest));

  	memset(&wr, 0, sizeof(wr));

	wr.wr_id = (uintptr_t)conn;
	wr.opcode = IBV_WR_SEND;
	wr.sg_list = &sge;
	wr.num_sge = 1;
	wr.send_flags = IBV_SEND_SIGNALED;

	sge.addr = (uintptr_t)client_conn->send_region;
	sge.length = sizeof(ReadRequest);
	sge.lkey = client_conn->send_mr->lkey;

	TEST_NZ(ibv_post_send(conn->qp, &wr, &bad_wr));
}

void post_receives4read(ReadRequest* request)
{
	struct Connection* conn = conn2datanodes[request->targetid];
	struct ibv_recv_wr wr, *bad_wr = NULL;
  	struct ibv_sge sge;
  	Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

  	wr.wr_id = (uintptr_t)conn;
  	wr.next = NULL;
  	wr.sg_list = &sge;
  	wr.num_sge = 1;

  	sge.addr = (uintptr_t)client_conn->recv_region;
  	sge.length = sizeof(ReadResponse);
  	sge.lkey = client_conn->recv_mr->lkey;

  	TEST_NZ(ibv_post_recv(conn->qp, &wr, &bad_wr));
}

void post_receives4write(WriteRequest* request)
{
	struct Connection* conn = conn2datanodes[request->targetid];
	struct ibv_recv_wr wr, *bad_wr = NULL;
  	struct ibv_sge sge;
  	Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

  	wr.wr_id = (uintptr_t)conn;
  	wr.next = NULL;
  	wr.sg_list = &sge;
  	wr.num_sge = 1;

  	sge.addr = (uintptr_t)client_conn->recv_region;
  	sge.length = sizeof(WriteResponse);
  	sge.lkey = client_conn->recv_mr->lkey;

  	TEST_NZ(ibv_post_recv(conn->qp, &wr, &bad_wr));
}


void init_block_loc()
{
	block_loc = (int**)malloc(sizeof(int*)*STRIPE_NUM);
	for (int i = 0; i < STRIPE_NUM; ++i) 
		block_loc[i] = (int*)malloc(sizeof(int)*(K+R));
	for (int i = 0; i < STRIPE_NUM; ++i)
	{
		int start = i%DATANODE_NUM;
		for (int j = 0; j < K+R; ++j)
		{
			int index = (start+j)%DATANODE_NUM;
			block_loc[i][j] = index; 
		}
	}

  // for (int i = 0; i < STRIPE_NUM; ++i)
  // {
  //   printf("%d,%d,%d,%d\n",block_loc[i][0],block_loc[i][1],block_loc[i][2],block_loc[i][3]);
  // }

}

//恢复reconvery_id节点中的数据
void reconstruction(int recovery_id,int blocknum){
	long time1,time2,time3;
	time1 = getcurrent_time();
	printf("reconstruction start\n");
	int fds[DATANODE_NUM];
	int v_lens[DATANODE_NUM];
	v_lens[recovery_id] = 2147483647;
	int min_v_len = 2147483647;
	UpdateMessage* message = (UpdateMessage*)malloc(sizeof(UpdateMessage));
	message->type=2;
	message->target_region = recovery_id;

	for (int i = 0; i < DATANODE_NUM; ++i)
	{
		if (i!=recovery_id)
		{
			fds[i] =connect_try(datanodes[i].tcp_ip,UPDATE_SOCKET_PORT);
			send_UpdateMessage(fds[i],message);
		} 
	}

	printf("send reconstruction to all DataNode\n");

	for (int i = 0; i < DATANODE_NUM; ++i)
	{
		if (i!=recovery_id)
		{
			recv_UpdateMessage(fds[i],message);
			v_lens[i] = message->v_len;
		} 
	}

	printf("recv reconstruction all v_lens\n");

	for (int i = 0; i < DATANODE_NUM; ++i){
		if (min_v_len>v_lens[i])
			min_v_len = v_lens[i];
	}

	message->type=2;
	message->v_len=min_v_len;
	printf("min_v_len:%d\n", min_v_len);
	for (int i = 0; i < DATANODE_NUM; ++i)
	{
		//printf("%d\n",i);
		if (i!=recovery_id)
		{
			send_UpdateMessage(fds[i],message);
		} 
	}
	printf("send consistency requests\n");
	for (int i = 0; i < DATANODE_NUM; ++i)
	{
		if (i!=recovery_id)
		{
			recv_UpdateMessage(fds[i],message);
		} 
	}
	printf("recv consistency responses\n");
	//保证一致性之后，开始恢复数据
	time2 = getcurrent_time();
	ReadRequest* request = (ReadRequest*)malloc(sizeof(ReadRequest));
	char** blocks = (char**)malloc(sizeof(char*)*K);
	for (int i = 0; i < K; ++i)
	{
		blocks[i] = (char*)malloc(sizeof(char)*BLK_SIZE);
	}
	for (int i = 0; i < blocknum; ++i){
		//1.拉取数据
		int count=0;
		for (int j = 0; j < DATANODE_NUM; ++j){
			if (j!=recovery_id){
				request->type=0;
				request->requestid=0;
				request->targetid=j;
				request->blockid = 1;
				exeReadtrace(request);
				count++;
			}	
			if (count==K) break;
		}
		//2.解码运算
		for (int j = 0; j < K; ++j)
		{
			for (int k = 0; k < BLK_SIZE; ++k)
			{
				blocks[j][k] = blocks[j][k]^0;
			}
		}
	}
	printf("reconstruction finish\n");
	time3 = getcurrent_time();
	printf("consistency phase time:%ld mius; fetch+decode time:%ld mius\n", time2-time1, time3-time2);
	printf("reconstruction_time:%ld mius\n",time3-time1);

}


void* conn_event(void* arg)
{
	struct rdma_cm_event *event = NULL;

	while (rdma_get_cm_event(event_channel, &event) == 0) {
    	struct rdma_cm_event event_copy;
    	memcpy(&event_copy, event, sizeof(*event));
    	rdma_ack_cm_event(event);
    	if (on_event(&event_copy))
      		break;
    }

  	rdma_destroy_event_channel(event_channel);
  	return NULL;
}

void* poll_cq(void *ctx)
{
  struct ibv_cq *cq;
  struct ibv_wc wc;

  while (1) {
    TEST_NZ(ibv_get_cq_event(s_ctx->comp_channel, &cq, &ctx));
    ibv_ack_cq_events(cq, 1);
    TEST_NZ(ibv_req_notify_cq(cq, 0));

    while (ibv_poll_cq(cq, 1, &wc))
      on_completion(&wc);
  }

  return NULL;
}


void build_context(struct ibv_context *verbs)
{
  if (s_ctx) {
    if (s_ctx->ctx != verbs)
      die("cannot handle events in more than one context.");

    return;
  }

  s_ctx = (struct Context *)malloc(sizeof(struct Context));

  s_ctx->ctx = verbs;

  TEST_Z(s_ctx->pd = ibv_alloc_pd(s_ctx->ctx));
  TEST_Z(s_ctx->comp_channel = ibv_create_comp_channel(s_ctx->ctx));
  TEST_Z(s_ctx->cq = ibv_create_cq(s_ctx->ctx, 32, NULL, s_ctx->comp_channel, 0)); /* cqe=10 is arbitrary */
  TEST_NZ(ibv_req_notify_cq(s_ctx->cq, 0));

  TEST_NZ(pthread_create(&s_ctx->cq_poller_thread, NULL, poll_cq, NULL));
}

void build_qp_attr(struct ibv_qp_init_attr *qp_attr)
{
  memset(qp_attr, 0, sizeof(*qp_attr));

  qp_attr->send_cq = s_ctx->cq;
  qp_attr->recv_cq = s_ctx->cq;
  qp_attr->qp_type = IBV_QPT_RC;

  qp_attr->cap.max_send_wr = 10;
  qp_attr->cap.max_recv_wr = 10;
  qp_attr->cap.max_send_sge = 1;
  qp_attr->cap.max_recv_sge = 1;
}

int on_addr_resolved(struct rdma_cm_id *id)
{
	printf("start address resolved.\n");
	struct sockaddr* addr =  rdma_get_peer_addr(id);
  	int peer_id = get_nodeid(addr,datanodes);
  	printf("on_addr_resolved:%d\n",peer_id);

  	struct ibv_qp_init_attr qp_attr;
  	build_context(id->verbs);
  	build_qp_attr(&qp_attr);
  	TEST_NZ(rdma_create_qp(id, s_ctx->pd, &qp_attr));

  	Connection* conn = (Connection*)malloc(sizeof(Connection));
  	conn->type = 0;
  	conn->id = id;
  	conn->qp = id->qp;
  	id->context = conn;

  	conn2datanodes[peer_id]=conn;
  	Client_Connection* client_conn = (Client_Connection*)malloc(sizeof(Client_Connection));
  	conn->conn_info = (void*)client_conn;

  	register_memory(conn);
  	TEST_NZ(rdma_resolve_route(id, TIMEOUT_IN_MS));
	printf("end address resolved.\n");
	return 0;
}

int on_route_resolved(struct rdma_cm_id *id)
{
	printf("route resolved.\n");
	struct rdma_conn_param cm_params;
  	build_params(&cm_params);
  	TEST_NZ(rdma_connect(id, &cm_params));
  	return 0;
}

int on_connection(struct rdma_cm_id *id)
{
	printf("client connection to datanode successfully\n");
	return 0;
}

int on_disconnect(struct rdma_cm_id *id)
{
	printf("peer disconnected.\n");
  	destroy_connection(id->context); 
  	return 0;
}

void destroy_connection(void *context)
{
    struct Connection *conn = (struct Connection *)context;
    rdma_destroy_qp(conn->id);
    Client_Connection* client_conn = (Client_Connection*)conn->conn_info;
    ibv_dereg_mr(client_conn->send_mr);
    ibv_dereg_mr(client_conn->recv_mr);
    free(client_conn->recv_region);
    free(client_conn->send_region);
    rdma_destroy_id(conn->id);
    free(conn);
}

int on_event(struct rdma_cm_event *event)
{
  int r = 0;
  printf("%s\n",rdma_event_str(event->event));
  if (event->event == RDMA_CM_EVENT_ADDR_RESOLVED)
    r = on_addr_resolved(event->id);
  else if (event->event == RDMA_CM_EVENT_ROUTE_RESOLVED)
    r = on_route_resolved(event->id);
  else if (event->event == RDMA_CM_EVENT_ESTABLISHED)
    r = on_connection(event->id->context);
  else if (event->event == RDMA_CM_EVENT_DISCONNECTED)
    r = on_disconnect(event->id);
  else
    die("on_event: unknown event.");
  return r;
}

void register_memory(Connection *conn)
{
  
    Client_Connection* client_conn = (Client_Connection*)conn->conn_info;
    client_conn->recv_region = (char*)malloc(SEND_RECV_BUFFER_SIZE);
    client_conn->send_region = (char*)malloc(SEND_RECV_BUFFER_SIZE);

    TEST_Z(client_conn->send_mr = ibv_reg_mr(
          s_ctx->pd,
          client_conn->send_region,
          SEND_RECV_BUFFER_SIZE,
          IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));

    TEST_Z(client_conn->recv_mr = ibv_reg_mr(
          s_ctx->pd,
          client_conn->recv_region,
          SEND_RECV_BUFFER_SIZE,
          IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));
}


void build_params(struct rdma_conn_param *params)
{
  memset(params, 0, sizeof(*params));
  params->initiator_depth = params->responder_resources = 1;
  params->rnr_retry_count = 7; /* infinite retry */
}

//request的处理逻辑
void on_completion(struct ibv_wc *wc)
{
  struct Connection *conn = (struct Connection *)(uintptr_t)wc->wr_id;
  Client_Connection* client_conn = (Client_Connection*)conn->conn_info;
  printf("start sloving the wc\n");
  if (wc->status != IBV_WC_SUCCESS)
    	die("on_completion: status is not IBV_WC_SUCCESS.");
  if (wc->opcode == IBV_WC_RECV){
    WriteResponse* response = (WriteResponse*)malloc(sizeof(WriteResponse));
    memcpy(response,client_conn->recv_region,sizeof(WriteResponse));

		pthread_mutex_lock(&mutex);
		flag=0;
		pthread_cond_signal(&cond);
		pthread_mutex_unlock(&mutex);
    
  }else if (wc->opcode == IBV_WC_SEND){
    printf("send completed successfully.\n");
  }else{
    die("on_completion: completion isn't a send or a receive.");
  }
}

void send_UpdateMessage(int fd,UpdateMessage* message)
{
    size_t needSend =sizeof(UpdateMessage);
    char *sendBuf=(char*)malloc(needSend);
    memcpy(sendBuf,message,needSend); 
    send_bytes(fd,sendBuf,needSend);
    //printf("111\n");
    free(sendBuf);
}

void recv_UpdateMessage(int fd,UpdateMessage* message)
{
    size_t needRecv =sizeof(UpdateMessage);
    char *recvBuf=(char *)malloc(needRecv);
    recv_bytes(fd,recvBuf,needRecv);
    memcpy(message,recvBuf,needRecv);  //反序列化
    free(recvBuf);
}
